package com.wsjj.yjh;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//如果输入的数据是sensor_1，只打印vc；如果输入的数据是sensor_2，既打印ts又打印vc。
public class TransFlatmap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 2L, 2),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
        );
        source.flatMap(new FlatMapFunction<WaterSensor, Long>() {
            @Override
            public void flatMap(WaterSensor value, Collector<Long> out) throws Exception {
                String id = value.getId();
                if(id.equals("sensor_1")){
                    out.collect(value.getVc().longValue());
//                    out.collect(Long.valueOf(value.getVc()));
                }
                if(id.equals("sensor_2")){
                    out.collect(value.getVc().longValue());
                    out.collect(value.getTs());
                }

            }
        }).print();
        env.execute();
    }
}
